package dse.messages.multiversion;

import java.util.*;

import dse.DataObject;
import dse.Experiment;
import dse.ExperimentRecorder;
import dse.Node;
import dse.Version;
import dse.messages.Message;
import dse.messages.MessagePurpose;
import dse.messages.MessageUtil;
import dse.messages.Transaction;
import dse.messages.singleversion.ShareDownloader;

public class GetUVNTransaction extends Transaction implements
		ShareDownloader, VersionReader{
	private Node requester;
	private DataObject desiredObject;
	
	private LinkedList<Node> residentNodes;
	private boolean failed;
	private TreeMap<Version, LinkedList<Node>> sharesDownloaded;
	private Hashtable<Node, LinkedList<Version>> versionsDownloaded;
	private int requestShareCount;
	private int validShareCount;
	private int maxFailure;
	private int successCount;
	private int failureCount;
	//private int preOffset;
	private int sharesToBeUpdate;
	//private boolean hasRequestMoreShares;
	//private boolean hasRequestPreviousShares;
	//private int round;

	public GetUVNTransaction(Experiment experiment,
			int id,
			Node requester,
			DataObject desiredObject,
			double time
			)
	{
		super(id, requester, experiment, time);
		
		this.requester = requester;
		this.desiredObject = desiredObject;
		this.failed = false;
		this.requestShareCount = 0;
		this.validShareCount = 0;
		this.maxFailure = experiment.modules().nodeEventScheduler().getEventsPerPeriod();
		this.successCount = 0;
		this.failureCount = 0;
		//this.preOffset = 0;
		this.sharesToBeUpdate = desiredObject.shareCount();
		if(isUpdate && isPartialUpdate)
		{
			this.sharesToBeUpdate = (int)(desiredObject.shareCount()*Math.random());
			//System.err.println("Partial Update");
		}
		//this.hasRequestMoreShares = false;
		//this.hasRequestPreviousShares = false;
		//this.round = 1;
		this.versionsDownloaded = null;
	}
	@Override
	public void start(ExperimentRecorder recorder) {
		// TODO Auto-generated method stub
		residentNodes = MessageUtil.getNodesByDht(experiment.network(), requester, desiredObject, experiment.modules().assigner());		
		
		if(isUpdate)
		{
			recorder.dataUpdateInitiated(super.id, requester, desiredObject, startTime);
			recorder.debug(this, "start", startTime);
			this.validShareCount = desiredObject.shareCount() - this.maxFailure;			
			Version version = new Version(requester.id(), requester.sequence());
			for(int i=0; (i<this.sharesToBeUpdate) && (i<residentNodes.size()); i++)	
				sendUpdateShareRequest(recorder, residentNodes.get(i), version, startTime);
		}
		else
		{
			recorder.dataRequestInitiated(super.id, requester, desiredObject, startTime);
			recorder.debug(this, "start", startTime);
			//this.requestShareCount = experiment.modules().shareRequestDecisionMaker().sharesToRequestFor(desiredObject);
						//+ this.maxFailure*2;
			//this.requestShareCount = this.requestShareCount>this.desiredObject.shareCount()?this.desiredObject.shareCount():this.requestShareCount;
			this.requestShareCount = desiredObject.requiredShareCount()+this.maxFailure;
			this.validShareCount = desiredObject.requiredShareCount(); //- this.maxFailure;
	
			for(int i=0; (i<requestShareCount)&&(i<residentNodes.size()); i++)
			{
				sendGetLatestAndVersionRequest(recorder, residentNodes.get(i), startTime);
			}
		}
	}

	@Override
	public boolean isDoneDownloadingShares() {
		// TODO Auto-generated method stub
		return failed || finished;
	}

	@Override
	public boolean isDoneUpdatingShares() {
		// TODO Auto-generated method stub
		return failed || finished;
	}
	
	

	@Override
	public void shareDownloadFailed(ExperimentRecorder recorder,
			DataObject dobj, int shareIndex, double timeLeft) {
		// TODO Auto-generated method stub
	}

	@Override
	public void shareDownloaded(ExperimentRecorder recorder, Node from,
			DataObject dobj, int shareIndex, double timeLeft) {
		// TODO Auto-generated method stub
	}

	@Override
	public void shareUpdateFailed(ExperimentRecorder recorder, DataObject dobj,
			int shareIndex, double timeLeft) {
		// TODO Auto-generated method stub
		if(isUpdate) {
			this.failureCount++;
			//if(this.processedUpdateMessage >= desiredObject.shareCount())
			if((successCount+failureCount) >= this.sharesToBeUpdate)
			{
				if(this.successCount >= validShareCount)
					transactionSucceed(recorder, timeLeft);
				else
					transactionFailed(recorder, timeLeft);
			}
		}
	}

	@Override
	public void shareUpdated(ExperimentRecorder recorder, Node to,
			DataObject dobj, int shareIndex, double timeLeft) {
		// TODO Auto-generated method stub
		if(isUpdate) {		
			this.successCount++;
			//if(this.processedUpdateMessage >= desiredObject.shareCount())
			if((successCount+failureCount) >= this.sharesToBeUpdate)
			{
				if(this.successCount >= validShareCount)
					transactionSucceed(recorder, timeLeft);
				else
					transactionFailed(recorder, timeLeft);
			}
		}
	}
	
	public void readVersion(ExperimentRecorder recorder, TreeSet<Version> versions, double time, Node from){}
	public void readVersion(ExperimentRecorder recorder, int seq, int index, double time){}

	public void readVersion(ExperimentRecorder recorder, Node from, Version version, double time){
		if(failed || finished)
			return;
		if(version == null){
			//failureCount++;
			transactionFailed(recorder, time);
			return;
		}
			
		//if(sharesDownloaded == null)
			//sharesDownloaded = new TreeMap<Version, LinkedList<Node>>();
		successCount++;
		//if(this.sharesDownloaded.get(version) == null)
		//{
			//this.sharesDownloaded.put(version, new LinkedList<Node>());
			//this.sharesDownloaded.get(version).add(from);
		//}
		//else
			//this.sharesDownloaded.get(version).add(from);
		
		if(successCount >= requestShareCount)
			//processVersion(recorder, time);
			transactionSucceed(recorder, time);
	}
	public void readVersion(ExperimentRecorder recorder, int seq, double time){}
	public void readVersion(ExperimentRecorder recorder, double time, LinkedList<Version> versions, Version version, Node from){
		if(finished || failed)
			return;
		if(sharesDownloaded == null)
			sharesDownloaded = new TreeMap<Version, LinkedList<Node>>();
		if(versionsDownloaded == null)
			versionsDownloaded = new Hashtable<Node, LinkedList<Version>>();
		
		if(version == null){
			failureCount++;
			//this.residentNodes.remove(from);
			if(failureCount > this.maxFailure){
				System.out.println("Failed number exceeds the maximum number");
				transactionFailed(recorder, time);
			}
			return;
		}
		
		successCount++;
		if(this.sharesDownloaded.get(version) == null)
		{
			this.sharesDownloaded.put(version, new LinkedList<Node>());
			this.sharesDownloaded.get(version).add(from);
		}
		else
			this.sharesDownloaded.get(version).add(from);
		
		versionsDownloaded.put(from, versions);
		recorder.shareHistorySize(versions.size());
		if(successCount+failureCount>= requestShareCount)
			processVersion(recorder, time);
	}
	
	private void processVersion(ExperimentRecorder recorder, double time)
	{
		if((sharesDownloaded==null) || (sharesDownloaded.size()==0))
		{
			transactionFailed(recorder, time);
			return;
		}
		
		Version maxVersion = sharesDownloaded.lastKey();
		if(maxVersion.getSequence()>requester.getSequence())
			requester.setSequence(maxVersion.getSequence());
		for(Version version : sharesDownloaded.keySet())
		{
			if(sharesDownloaded.get(version).size() == requestShareCount)
			{
				transactionSucceed(recorder, time);
				return;
			}
			else if(sharesDownloaded.get(version).size() >= validShareCount)
			{
				for(Version ver : sharesDownloaded.keySet())
				{
					if(version.compareTo(ver) != 0)
					{
						for(Node node : sharesDownloaded.get(ver))
						{
							sendUpdateShareRequest(recorder, node, version, time);
						}
					}
				}
				transactionSucceed(recorder, time);	
				return;
			}
		}
		
		//int srf = experiment.modules().shareRequestDecisionMaker().sharesToRequestFor(desiredObject)
				//+ this.maxFailure*2;
		/*if(this.hasRequestPreviousShares)
			recorder.requestPreviousShares(false);
		else{
			this.hasRequestPreviousShares = true;
			recorder.requestPreviousShares(true);
		}*/
		failureCount = 0;
		successCount = 0;
		this.sharesDownloaded.clear();
		processVersions(recorder, time);
		//preOffset++;
		//this.round++;
		//for(int i=0; (i<this.requestShareCount)&&(i<residentNodes.size()); i++)
			//sendGetPreviousShareRequest(recorder, preOffset, residentNodes.get(i), time);
	}
	
	private void processVersions(ExperimentRecorder recorder, double time){
		if((versionsDownloaded==null) || (versionsDownloaded.size()==0))
		{
			transactionFailed(recorder, time);
			return;
		}
		
		Hashtable<Version, LinkedList<Node>> versions = new Hashtable<Version, LinkedList<Node>>();
		int i = 0;
		while(true){
			LinkedList<Node> nodes = new LinkedList<Node>();
			for(Node node : versionsDownloaded.keySet()){
				if(i >= this.versionsDownloaded.get(node).size()){
					nodes.add(node);
					continue;
				}				
				
				Version v = this.versionsDownloaded.get(node).get(i);
				
				if(versions.get(v) == null){
					versions.put(v, new LinkedList<Node>());
					versions.get(v).add(node);
				}
				else
					versions.get(v).add(node);
				
				if(versions.get(v).size() >= this.desiredObject.requiredShareCount()){
					LinkedList<Node> availNodes = versions.get(v);
					this.requestShareCount = desiredObject.requiredShareCount();
					
					for(int j=0; j<this.requestShareCount; j++)
						SendRequestDesignatedShareRequest(recorder, v, availNodes.get(j), time);
					return;
				}
			}
			while(nodes.size() > 0){
				this.versionsDownloaded.remove(nodes.pollFirst());
			}
			
			if(versionsDownloaded.size() == 0){
				System.err.println("No data available");
				this.transactionFailed(recorder, time);
				return;
			}
			i++;
		}
	}
	
	private void sendUpdateShareRequest(ExperimentRecorder recorder, Node node, Version version, Double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createShareUpdateV3Message(requester, node, desiredObject, 
				version, MessagePurpose.DataObjectUpdate, this);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	private void sendGetLatestAndVersionRequest(ExperimentRecorder recorder, Node node, double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createRequestLatestAndVersionV2Message(requester, node, desiredObject, 
				MessagePurpose.DataObjectRequest, this);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	private void SendRequestDesignatedShareRequest(ExperimentRecorder recorder, Version version, Node node, double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createRequestDesignatedShareMessage(requester, node, desiredObject, version,
				MessagePurpose.DataObjectRequest, this);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	private void transactionSucceed(ExperimentRecorder recorder, double time)
	{
		finished = true;
		failed  = false;
		//System.out.println("Data ID : " + desiredObject.id() + "Version : " + maxVersion.getSequence() + " , " + maxVersion.getClientID());
		if(isUpdate)
		{
			recorder.debug(
					"GetDhtDataSharesWithVersionTransaction trans#" + super.id + 
					" is finshed!!!"
					);
			recorder.dataUpdateFinished(super.id, requester, desiredObject, true, time);
		}
		else
		{
			recorder.debug(
					"GetDhtDataSharesWithVersionTransaction trans#" + super.id + 
					" is finshed!!!"
					);
			recorder.dataRequestFinished(super.id, requester, desiredObject, true, this.residentNodes.get(this.validShareCount-1),
					false, time);
			for(int i=0; (i<requestShareCount)&&(i<residentNodes.size()); i++)
			{
				Message m = super.createReadCompleteMessage(requester, this.residentNodes.get(i), desiredObject, MessagePurpose.DataObjectRequest, this, true);
				recorder.debug(this, "new msg: " + m, time);
				experiment.messageQueue().enqueueMessage(m, time);
			}
		}
	}
	
	private void transactionFailed (ExperimentRecorder recorder, double time)
	{
		recorder.debug(this, "transFailed", time);
		
		finished = true;
		failed = true;
		if(isUpdate){
			recorder.dataUpdateFinished(super.id, requester, desiredObject, false, time);
		}
		else{
			recorder.dataRequestFinished(super.id, requester, desiredObject, false, null,
				false, time);
			for(int i=0; (i<requestShareCount)&&(i<residentNodes.size()); i++)
			{
				Message m = super.createReadCompleteMessage(requester, this.residentNodes.get(i), desiredObject, MessagePurpose.DataObjectRequest, this, true);
				recorder.debug(this, "new msg: " + m, time);
				experiment.messageQueue().enqueueMessage(m, time);
			}
		}
	}
	public Transaction getTransaction(){
		return this;
	}
}
